package com.yanggu.spark.core.demand

import org.apache.spark.{SparkConf, SparkContext}

//练习
//agent.log：时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
object AgentLogDemo {

  def main(args: Array[String]): Unit = {

    //1. 创建sparkConf配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")

    //2. 创建spark上下文对象
    val sparkContext = new SparkContext(sparkConf)

    //3. 读取文件, 创建RDD
    //时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
    //1516609143867 6 7 64 16
    //统计出每一个省份每个广告被点击数量排行的Top3
    val logRDD = sparkContext.textFile("input/agent.log")

    //4. 需求实现
    val result = logRDD
      //映射成 => ((省份, 广告), 1)
      .map(line => {
        val array = line.split(" ")
        ((array(1), array(4)), 1)
      })
      //规约成 => ((省份, 广告), sum)
      .reduceByKey(_ + _)
      //转换结构 => (省份, (广告, sum))
      .map {
        case ((provinceId, adId), count) =>
          (provinceId, (adId, count))
      }
      //根据省份分组 => (省份, Iterable(广告, sum))
      .groupByKey()
      //得到所有的values
      .mapValues(
        //转换成List
        _.toList
          //sortBy可以根据元祖的顺序进行排序
          .sortBy(t => (-t._2, -t._1.toInt))
          //获取前3个
          .take(3)
      )
      .collect

    //5. 打印
    result.foreach {
      case (provinceId, list) =>
        list.foreach {
          case (adId, count) =>
            println(s"省份的id: $provinceId, 广告的id: $adId, 点击次数: $count")
        }
    }

    //6. 释放资源
    sparkContext.stop()
  }

}
